package SparkGraphXInAction

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx._

/**
  * 这里还是为每个顶点标记上离它最远的根顶点的距离。
  * 但是使用的技术不再是aggragateMessages,而是Pregel API.
  * Created by Administrator on 2017/4/23 0023.
  * 算法的最终结果，我们希望为每个顶点标记上离它最远的根顶点的距离。
  * 假设图不存在环（环表示当沿着边从一个顶点出发，最终又回到同一个顶点）。
  * 处理包含环的图通常会增加算法的复杂性，我们稍后会介绍一些处理环的策略。
  */
object DistanceFromRootVertex2 {


  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    val myVertices = sc.makeRDD(Array((1L, "Ann"),(2L, "Bill"), (3L, "Charles"), (4L, "Diane"), (5L, "Went to gym this morning")))
    val myEdges = sc.makeRDD(Array(Edge(1L, 2L, "is-friends-with"), Edge(2L, 3L, "is-friends-with"),
      Edge(3L, 4L, "is-friends-with"), Edge(4L, 5L, "Likes-status"), Edge(3L, 5L, "Wrote-status")))
    val myGraph = Graph(myVertices, myEdges)

    val g = Pregel(myGraph.mapVertices((vid,vd) => 0), 0, activeDirection = EdgeDirection.Out)(
      (id:VertexId, vd:Int, a:Int) => math.max(vd, a),(et:EdgeTriplet[Int,String]) => Iterator((et.dstId,et.srcAttr+1)), (a:Int, b:Int) => math.max(a,b))
//    println(g.asInstanceOf[AnyRef].getClass.getSimpleName)
//    println(myGraph.asInstanceOf[AnyRef].getClass.getSimpleName)
    val lll = g.vertices.collect
    for(vertice <- lll){
      println(vertice)
    }
    /*结果：
    (4,3)
    (1,0)
    (3,2)
    (5,4)
    (2,1)
     */
  }
}
